package com.lagou.zookeeper;

import com.lagou.client.RPCConsumer;
import com.lagou.service.IUserService;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.net.ConnectException;
import java.util.*;
import java.util.function.ToLongFunction;
import java.util.stream.Collectors;

/**
 * @program: stage-3-module-2
 * @description:
 * @author: zhangxiaoyang
 * @date: 2020-08-23 22:36
 **/
public class GetNodeData implements Watcher {
    private static final String PROVIDER_NAME = "UserService#sayHello#";


    public static ZooKeeper zooKeeper;

    private RPCConsumer rpcConsumer;


    public static Map<String, Long> addressMap = new HashMap<>();


    public static void init() throws IOException {
        zooKeeper = new ZooKeeper(ZkConstant.ZK_SERVER + ":" + ZkConstant.ZK_PORT, 50000, new GetNodeData());
        System.out.println("消费者客户端zk连接成功！！！");
    }

    @Override
    public void process(WatchedEvent watchedEvent) {
        if (watchedEvent.getType() == Event.EventType.NodeChildrenChanged) {
            try {
                List<String> children = getChildren();
                for (int i = 0; i < children.size(); i++) {
                    if (addNodeData(children.get(i))) {
                        String[] arr = children.get(i)
                                               .split(":");
                        try {
                            this.connect(arr[0], Integer.valueOf(arr[1]));
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                    }
                }
                for (Map.Entry<String, Long> entry : addressMap.entrySet()) {
                    Stat stat = zooKeeper.exists(ZkConstant.ROOT_NODE + "/" + entry.getKey(), false);
                    if (stat == null) {
                        rpcConsumer.disconnect(entry.getKey());
                    }
                }
            } catch (UnsupportedEncodingException e) {
                e.printStackTrace();
            } catch (KeeperException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        if (watchedEvent.getState() == Event.KeeperState.SyncConnected) {
            try {
                List<String> children = getChildren();
                for (String child : children) {
                    if (addNodeData(child)) {
                        String[] arr = child.split(":");
                        try {
                            this.connect(arr[0], Integer.valueOf(arr[1]));
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                    }
                }
            } catch (KeeperException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (UnsupportedEncodingException e) {
                e.printStackTrace();
            }
        }
        if (watchedEvent.getType() == Event.EventType.NodeDeleted) {
            try {
                for (Map.Entry<String, Long> entry : addressMap.entrySet()) {
                    Stat stat = zooKeeper.exists(ZkConstant.ROOT_NODE + "/" + entry.getKey(), false);
                    if (stat == null) {
                        rpcConsumer.disconnect(entry.getKey());
                    }
                }
            } catch (KeeperException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

    }

    private List<String> getChildren() throws KeeperException, InterruptedException {
        List<String> children = zooKeeper.getChildren(ZkConstant.ROOT_NODE, true);
        return children;
    }

    private boolean addNodeData(String child) throws KeeperException, InterruptedException, UnsupportedEncodingException {
        byte[] bytes = zooKeeper.getData(ZkConstant.ROOT_NODE + "/" + child, true, null);
        return addressMap.put(child, Long.valueOf(new String(bytes, "UTF-8"))) == null;
    }


    public void connect(String ip, int port) throws IOException, InterruptedException, KeeperException {
        System.out.println("和服务端建立连接----------------------");
        rpcConsumer = new RPCConsumer();
        //初始化zk
        GetNodeData.init();
        //和新加入的节点建立连接
        rpcConsumer.initClient(ip, port);

        //1.创建代理对象
        IUserService service = (IUserService) rpcConsumer.createProxy(IUserService.class, PROVIDER_NAME);
        //2.循环给服务器写数据
        while (true) {
            //负载均衡
            String address = loadBalance();
            String[] arr = address.split(":");
            //初始化客户端
            try {
                rpcConsumer.initClient(arr[0], Integer.valueOf(arr[1]));
                String result = service.sayHello("are you ok !!");
                System.out.println(result);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (KeeperException e) {
                e.printStackTrace();
            } catch (Exception e) {
                e.printStackTrace();
                rpcConsumer.disconnect(address);
            }
            Thread.sleep(2000);
        }
    }

    private String loadBalance() throws KeeperException, InterruptedException, UnsupportedEncodingException {
        List<String> children = getChildren();
        Map<String, Long> dataMap = getData(children);
        int minIndex = new Random().nextInt(children.size());
        //负载均衡
        for (int i = 1; i < children.size(); i++) {
            if (children.size() == 1) {
                minIndex = i;
                return children.get(minIndex);
            } else {
                if (dataMap.get(children.get(i - 1)) < dataMap.get(children.get(i))) {
                    minIndex = i - 1;
                } else if (dataMap.get(children.get(i - 1)) > dataMap.get(children.get(i))) {
                    minIndex = i;
                } else {
                    return children.get(minIndex);
                }
            }
        }
        return children.get(minIndex);
    }

    private Map<String, Long> getData(List<String> children) throws UnsupportedEncodingException, KeeperException, InterruptedException {
        Map<String, Long> dataMap = new HashMap<>();
        for (String child : children) {
            byte[] bytes = GetNodeData.zooKeeper.getData(ZkConstant.ROOT_NODE + "/" + child, true, null);
            System.out.println(ZkConstant.ROOT_NODE + "/" + child + "节点数据=====" + new String(bytes, "UTF-8"));
            dataMap.put(child, Long.valueOf(new String(bytes, "UTF-8")));
        }
        return dataMap;
    }


}
